package com.mimaxueyuan.disruptor.demo1;

import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

/**
 * @author: Kevin
 * @官网: 	www.mimaxueyuan.com
 * @Q Q群:	660567408
 * @Email:	mimaxueyuan@163.com
 * [每天进步一点点、人生带来大改变...]
 * [本代码对应视频地址:http://study.163.com/course/introduction/1004573050.htm]
 */
public class LongEventMain3 {

	@SuppressWarnings("unchecked")
	public static void main(String[] args) throws InterruptedException {
		// 初始化线程池-用户执行Consumer
		//Executor threadFactory = Executors.newCachedThreadPool();
		ThreadFactory threadFactory = Executors.defaultThreadFactory();

		// 初始化EventFactory
		LongEventFactory factory = new LongEventFactory();

		// 初始化RingBuffer的大小,必须是2的指数
		int bufferSize = 1024;

		// 初始化Disruptor
		//默认配置
		//Disruptor<LongEvent> disruptor = new Disruptor<LongEvent>(factory, bufferSize, threadFactory);
		//多线程配置
		//Disruptor<LongEvent> disruptor = new Disruptor<LongEvent>(factory, bufferSize, threadFactory,ProducerType.MULTI,new BlockingWaitStrategy());
		//单线程配置
		Disruptor<LongEvent> disruptor = new Disruptor<LongEvent>(factory, bufferSize, threadFactory,ProducerType.SINGLE,new BlockingWaitStrategy());

		// 指定事件处理器
		disruptor.handleEventsWith(new LongEventHandler());

		// 开启Disruptor,开启所有线程
		disruptor.start();

		// 获取RingBuffer
		RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();

		//采用固定模式向RingBuffer中添加事件
		final LongEventProducer producer = new LongEventProducer(ringBuffer);

		//10线程并发下的Disruptor
		final CyclicBarrier cb = new CyclicBarrier(6,()->{
			System.out.println("--------完成任务");
		});
		ExecutorService executor1 = Executors.newCachedThreadPool();
		for (long i = 0; i<10; i++) {
			final long vv = i;
			executor1.submit(new Runnable() {
				public void run() {
					try {
						System.out.println(vv+"is ready..");
						cb.await();
						producer.onData(vv);
					}catch(Exception e) {
						e.printStackTrace();
					}
				}
			});
		}

		//disruptor.shutdown();
	}

}
